﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Common.Entitys.CommonLogs;
using Common.Entitys.Messages;
using Common.Utlis;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
using MQTTnet.Client.Publishing;
using MQTTnet.Client.Receiving;
using MQTTnet.Client.Subscribing;
using MQTTnet.Exceptions;
using MQTTnet.Packets;
using MQTTnet.Protocol;


namespace Common.Utils.MqttHelp
{
    public class MqttClientHelp
    {
        private MqttClientOptions options;

        private MqttClient mqttClient = null;

        private readonly object m_LockDictTopicInfos = new object();

        /// <summary>
        /// 需要订阅的消息列表
        /// </summary>
        private Dictionary<string, MqttQualityOfServiceLevel> m_DictTopicInfos =
            new Dictionary<string, MqttQualityOfServiceLevel>();

        /// <summary>
        /// 断线时的消息
        /// </summary>
        private ConcurrentQueue<IPCMsg> m_QueIPCMsg = new ConcurrentQueue<IPCMsg>();


        public MqttClientHelp()
        {
            MessageAggregator<IPCMsg>.Subscribe(OnIPCMsg);
            MessageAggregator<MqttTopicFilter>.Subscribe(OnTopicAndLevelMsg);
        }

        /// <summary>
        ///添加话题(订阅消息)
        /// </summary>
        /// <param name="topicAndLevelMsg"></param>
        private async void OnTopicAndLevelMsg(MqttTopicFilter topicAndLevelMsg)
        {
            //断线暂存逻辑
            if (mqttClient == null || mqttClient.IsConnected == false)
            {
                lock (m_LockDictTopicInfos)
                {
                    //无则插入 有则更新
                    if (!m_DictTopicInfos.TryAdd(topicAndLevelMsg.Topic, topicAndLevelMsg.QualityOfServiceLevel))
                    {
                        m_DictTopicInfos[topicAndLevelMsg.Topic] = topicAndLevelMsg.QualityOfServiceLevel;
                    }
                }
                return;
            }

            //订阅逻辑
            try
            {
                
                MqttClientSubscribeResult mqttClientSubscribeResult = await mqttClient.SubscribeAsync(topicAndLevelMsg.Topic, topicAndLevelMsg.QualityOfServiceLevel);
                foreach (MqttClientSubscribeResultItem mqttClientSubscribeResultItem in mqttClientSubscribeResult.Items)
                {
                    switch (mqttClientSubscribeResultItem.ResultCode)
                    {
                        case MqttClientSubscribeResultCode.GrantedQoS0:
                        case MqttClientSubscribeResultCode.GrantedQoS1:
                        case MqttClientSubscribeResultCode.GrantedQoS2:
                            //成功了
                            //lock (m_LockDictTopicInfos)
                            //{
                            //    m_DictTopicInfos.Remove(topicAndLevelMsg.Topic);
                            //}
                            break;
                        default:
                            //未知在何种情况会出现这种结果(事实上的失败,日志记录)
                            ErrorLogEntity errorLogEntity = new ErrorLogEntity($"Mqtt订阅消息 {topicAndLevelMsg.Topic} 的返回值为 {mqttClientSubscribeResultItem.ResultCode} ,请检查用户名密码是否正确");
                            LogHelp.AddLog(errorLogEntity);
                            break;
                    }
                }
            }
            catch (MqttCommunicationException mqttCommunicationException)
            {
                WarnLogEntity warnLogEntity = new WarnLogEntity($"Mqtt订阅消息 {topicAndLevelMsg.Topic} 时 连接出现问题,请检查网络状态 {mqttCommunicationException.Message} ");
                LogHelp.AddLog(warnLogEntity);
                //连接出现问题,等待重连即可
                OnTopicAndLevelMsg(topicAndLevelMsg);
            }
            catch (Exception e)
            {
                //未知异常
                ErrorLogEntity errorLogEntity = new ErrorLogEntity($"Mqtt订阅消息 {topicAndLevelMsg.Topic} 时 出现未知异常  ", e);
                LogHelp.AddLog(errorLogEntity);
                OnTopicAndLevelMsg(topicAndLevelMsg);
            }
        }

        /// <summary>
        /// 批量添加话题
        /// </summary>
        /// <param name="lisMqttTopicFilters"></param>
        private async void OnTopicAndLevelMsg(List<MqttTopicFilter> lisMqttTopicFilters)
        {
            if (lisMqttTopicFilters.Count==0)
            {
                return;
            }
            try
            {
                //订阅 这时候可能会突然掉线
                MqttClientSubscribeResult mqttClientSubscribeResult = await mqttClient.SubscribeAsync(lisMqttTopicFilters.ToArray());
                foreach (MqttClientSubscribeResultItem mqttClientSubscribeResultItem in mqttClientSubscribeResult.Items)
                {
                    switch (mqttClientSubscribeResultItem.ResultCode)
                    {
                        case MqttClientSubscribeResultCode.GrantedQoS0:
                        case MqttClientSubscribeResultCode.GrantedQoS1:
                        case MqttClientSubscribeResultCode.GrantedQoS2:
                            //lock (m_LockDictTopicInfos)
                            //{
                            //    m_DictTopicInfos.Remove(mqttClientSubscribeResultItem.TopicFilter.Topic);
                            //}
                            break;
                        default:
                            //未知在何种情况会出现这种结果(事实上的失败,日志记录)
                            ErrorLogEntity errorLogEntity = new ErrorLogEntity($"Mqtt批量订阅消息 {mqttClientSubscribeResultItem.TopicFilter.Topic} 的返回值为 {mqttClientSubscribeResultItem.ResultCode} ,请检查用户名密码是否正确");
                            LogHelp.AddLog(errorLogEntity);
                            break;
                    }
                }
            }
            catch (MqttCommunicationException mqttCommunicationException)
            {
                WarnLogEntity warnLogEntity = new WarnLogEntity($"Mqtt批量订阅消息时 连接出现问题,请检查网络状态 {mqttCommunicationException.Message} ");
                LogHelp.AddLog(warnLogEntity);
                //连接出现问题,等待重连即可
                OnTopicAndLevelMsg(lisMqttTopicFilters);
            }
            catch (Exception e)
            {
                //未知异常
                ErrorLogEntity errorLogEntity = new ErrorLogEntity($"Mqtt订阅消息时 出现未知异常  ", e);
                LogHelp.AddLog(errorLogEntity);
                OnTopicAndLevelMsg(lisMqttTopicFilters);
            }

        }

        /// <summary>
        /// 接收IPC的消息
        /// </summary>
        /// <param name="ipcMsg"></param>
        private async void OnIPCMsg(IPCMsg ipcMsg)
        {
            if (mqttClient == null || (mqttClient.IsConnected == false))
            {
                switch (ipcMsg.MqttQualityOfServiceLevel)
                {
                    //不重要的消息丢掉
                    case MqttQualityOfServiceLevel.AtMostOnce:
                        return;
                    case MqttQualityOfServiceLevel.AtLeastOnce:
                    case MqttQualityOfServiceLevel.ExactlyOnce:
                        m_QueIPCMsg.Enqueue(ipcMsg);
                        break;
                }
                return;
            }
            //载入ClientId
            ipcMsg.ClientId = options.ClientId;

            //发布消息
            Byte[] payload = SerializeHelp.SerializeToByteArray(ipcMsg);
            //通常，如果发布者向主题发布消息，并且没有人订阅该主题，则该消息将被代理放弃。
            //但是，发布者可以通过设置保留的消息标志来告诉代理保留该主题的最后一条消息。
            //这可能非常有用，例如，如果您的传感器仅在更改时发布其状态，例如门传感器。如果新订户订阅此状态会怎样
            var message = new MqttApplicationMessage()
            {
                Topic = ipcMsg.TypeGuid.ToString("N"),
                Payload = payload,
                QualityOfServiceLevel = ipcMsg.MqttQualityOfServiceLevel,
                Retain = false
            };

            try
            {
                MqttClientPublishResult mqttClientPublishResult = await mqttClient.PublishAsync(message);
                if (message.QualityOfServiceLevel == MqttQualityOfServiceLevel.AtMostOnce)
                {
                    return;
                }

                if (mqttClientPublishResult.ReasonCode == MqttClientPublishReasonCode.Success)
                {
                    return;
                }

                ErrorLogEntity errorLogEntity = new ErrorLogEntity($"Mqtt发布消息 {message.Topic} 返回值异常为 {mqttClientPublishResult.ReasonCode} 请检查用户名密码是否正确");
                LogHelp.AddLog(errorLogEntity);
            }
            catch (MqttCommunicationException mqttCommunicationException)
            {
                WarnLogEntity warnLogEntity = new WarnLogEntity($"Mqtt发布消息 {message.Topic} 出现连接异常,请检查网络状态,已再次发送  异常详情为 {mqttCommunicationException.Message} ");
                LogHelp.AddLog(warnLogEntity);
                OnIPCMsg(ipcMsg);
            }
            catch (Exception e)
            {
                ErrorLogEntity errorLogEntity = new ErrorLogEntity($"Mqtt发布消息 {message.Topic} 出现未知异常 {e.Message} ");
                LogHelp.AddLog(errorLogEntity);
                OnIPCMsg(ipcMsg);
            }

        }

        public async Task Init(string clientId = "TableProject", string ip = "192.168.1.16", int port = 8086,  string userName = "Zane", string password = "123456")
        {
            try
            {
                var mqttFactory = new MqttFactory();
                options = new MqttClientOptions
                {
                    ClientId = clientId,
                    ProtocolVersion = MQTTnet.Formatter.MqttProtocolVersion.V311,
                    CleanSession = false, 
                    //通常，如果发布者向主题发布消息，并且没有人订阅该主题，则该消息将被代理放弃。
                    //但是，发布者可以通过设置保留的消息标志来告诉代理保留该主题的最后一条消息。
                    //这可能非常有用，例如，如果您的传感器仅在更改时发布其状态，例如门传感器。如果新订户订阅此状态会怎样
                    ChannelOptions = new MqttClientTcpOptions
                    {
                        Server = ip,
                        Port = port
                    },
                    WillDelayInterval = 10,
                    WillMessage = new MqttApplicationMessage()
                    {
                        Topic = $"LastWill/{clientId}",
                        Payload = Encoding.UTF8.GetBytes("I Lost the connection!"),
                        QualityOfServiceLevel = MqttQualityOfServiceLevel.ExactlyOnce
                    },
                    UserProperties = new List<MqttUserProperty>(){new MqttUserProperty("123","456")}
                };
                if (options.ChannelOptions == null)
                {
                    throw new InvalidOperationException();
                }


                options.Credentials = new MqttClientCredentials
                {
                    Username = userName,
                    Password = Encoding.UTF8.GetBytes(password),
                };

                options.CleanSession = true;
                options.KeepAlivePeriod = TimeSpan.FromSeconds(5);

                mqttClient = mqttFactory.CreateMqttClient() as MqttClient;
                mqttClient.ConnectedHandler = new MqttClientConnectedHandlerDelegate(OnMqttClientConnected);
                mqttClient.DisconnectedHandler = new MqttClientDisconnectedHandlerDelegate(OnMqttClientDisConnected);
                mqttClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(OnSubscriberMessageReceived);
                await mqttClient.ConnectAsync(options);
            }
            catch (MqttCommunicationException mqttCommunicationException)
            {
                WarnLogEntity warnLogEntity = new WarnLogEntity($"首次连接Mqtt服务器时失败!,1秒后自动尝试重连,请检查网络状态 错误信息 {mqttCommunicationException.Message} ");
                LogHelp.AddLog(warnLogEntity);
            }
            catch (Exception ex)
            {
                ErrorLogEntity errorLog = new ErrorLogEntity("连接到MQTT服务器失败!出现未知异常", ex);
                LogHelp.AddLog(errorLog);
            }

        }



        private  void OnMqttClientConnected(MqttClientConnectedEventArgs obj)
        {
            LogHelp.AddLog<InitLogEntity>("已连接到MQTT服务器");
            //处理断线期间的重要消息
            Task.Factory.StartNew(() =>
            {
                while (true)
                {
                    bool haveValue = m_QueIPCMsg.TryDequeue(out IPCMsg ipcMsg);
                    if (haveValue == false)
                    {
                        if (m_QueIPCMsg.IsEmpty)
                        {
                            break;
                        }
                        continue;
                    }
                    OnIPCMsg(ipcMsg);
                }
            });

            List<MqttTopicFilter> lisMqttTopicFilters = new List<MqttTopicFilter>();
            lisMqttTopicFilters.Add(new MqttTopicFilter(){Topic = "W"});
            lock (m_LockDictTopicInfos)
            {
                foreach (KeyValuePair<string, MqttQualityOfServiceLevel> mqttQualityOfServiceLevel in m_DictTopicInfos)
                {
                    MqttTopicFilter mqttTopicFilter = new MqttTopicFilter()
                    {
                        Topic = mqttQualityOfServiceLevel.Key,
                        QualityOfServiceLevel = mqttQualityOfServiceLevel.Value
                    };
                    lisMqttTopicFilters.Add(mqttTopicFilter);
                }
            }

            //批量添加话题(一次性添加容易成功)
            OnTopicAndLevelMsg(lisMqttTopicFilters);
            //发布消息
            MqttConnectMsg mqttConnectMsg = new MqttConnectMsg();
            mqttConnectMsg.ClientID = options.ClientId;
            MessageAggregator<MqttConnectMsg>.Publish(mqttConnectMsg);
        }

        private async void OnMqttClientDisConnected(MqttClientDisconnectedEventArgs e)
        {
            LogHelp.AddLog<WarnLogEntity>("已断开MQTT连接!");
            try
            {
                Thread.Sleep(TimeSpan.FromSeconds(5));
                await mqttClient.ReconnectAsync();
            }
            catch (MqttCommunicationException mqttCommunicationException)
            {
                WarnLogEntity warnLogEntity = new WarnLogEntity($"连接Mqtt服务器时失败!,1秒后自动尝试重连,请检查网络状态 错误信息 {mqttCommunicationException.Message} ");
                LogHelp.AddLog(warnLogEntity);
            }
            catch (Exception ex)
            {
                ErrorLogEntity errorLog = new ErrorLogEntity("连接到MQTT服务器出现未知异常！", ex);
                LogHelp.AddLog(errorLog);
            }
        }

        private void OnSubscriberMessageReceived(MqttApplicationMessageReceivedEventArgs e)
        {
            //拒收自己的消息(内部消息内部通道,)
            IPCMsg ipcMsg = SerializeHelp.ByteToObj(e.ApplicationMessage.Payload) as IPCMsg;
            if (ipcMsg.ClientId == options.ClientId)
            {
                return;
            }
            MessageAggregator.Publish(ipcMsg.Obj);
        }
    }
}
